package com.hhq.cloud.watch;

import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock;
import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
import io.fabric8.kubernetes.client.informers.SharedInformerFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;

/**
 * 动态监听
 */
@Component
@Slf4j
public class DynamicNamespaceWatcher  {
    private static volatile boolean isLeader = false;

    @Resource
    private KubernetesClient kubernetesClient;

    @Resource
    ThreadPoolExecutor KubernetesThreadPoolExecutor;

    @PostConstruct
    public void init() {
        String leaseNamespace = "default";
        String leaseName = "dynamic-leader-election";
        String identity = "node-" + System.currentTimeMillis();  // Unique identity for this node
        LeaseLock leaseLock = new LeaseLock(leaseNamespace, leaseName, identity);
        // LeaderElectionConfig 配置
        LeaderElectionConfig leaderConfig = new LeaderElectionConfig(
                leaseLock, // lock (leaseClient)
                Duration.ofSeconds(10),  // 租约持续的时间
                Duration.ofSeconds(15),  // 更新租约的最大时长
                Duration.ofSeconds(5),   // 重试选举的时间间隔
                new LeaderCallbacks(    //包含了当获得领导权、失去领导权和发生错误时的回调方法
                        // 当节点被选为领导者时
                        () -> {
                            isLeader = true;
                            log.info("[LEADER] Gained leadership: {}" , identity);
                            watchPods();
                        },
                        // 当节点失去领导权时
                        () -> {
                            isLeader = false;
                            log.info("[LEADER] Lost leadership: {}" ,identity);
                            stopWatchPods();
                        },
                        // 当选举产生新的领导者时
                        (newLeaderIdentity) -> {
                            log.info("[LEADER] New leader elected: {}" ,newLeaderIdentity);
                            retryWatchPods();
                        }
                ),
                true,  // 示当选举取消时，当前领导者会自动释放领导权。这样可以保证选举过程中的领导权不会被意外保留。
                leaseName // lease name
        );

        //   创建领导者选举
        LeaderElector elector = new LeaderElector(kubernetesClient,leaderConfig,KubernetesThreadPoolExecutor);
        // 启动 LeaderElector 进行领导者选举
        elector.start();
    }


    private  void watchPods() {
        // 在这个地方进行Pod监听
        SharedInformerFactory informerFactory = kubernetesClient.informers();
        informerFactory.sharedIndexInformerFor(Pod.class, 10 * 60 * 1000).addEventHandler(new ResourceEventHandler<Pod>() {
            @Override
            public void onAdd(Pod obj) {

            }

            @Override
            public void onUpdate(Pod oldObj, Pod newObj) {

            }

            @Override
            public void onDelete(Pod obj, boolean deletedFinalStateUnknown) {

            }
        });
        // 启动监听
        CompletableFuture<Void> future = (CompletableFuture<Void>) informerFactory.startAllRegisteredInformers();
        future.whenComplete((b,e)->{
            log.info("Connection interrupted, attempting to reconnect...");
            // 如果发生中断，可以尝试重新建立连接
            informerFactory.startAllRegisteredInformers();
        });
    }

    /**
     * 停止监听
     */
    private void stopWatchPods(){
        SharedInformerFactory informerFactory = kubernetesClient.informers();
        informerFactory.stopAllRegisteredInformers();
    }
    // 手动重连
    private  void retryWatchPods() {
        log.info("Attempting to reconnect...");
        // 可以加入延时等策略，避免频繁重试
        try {
            Thread.sleep(5000);  // 等待5秒钟重试
            watchPods();  // 重新开始监听
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @PreDestroy
    public void destory(){
        stopWatchPods();
    }
}
